package com.navinfo.data.esper.cep.linkid;


import com.espertech.esper.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class LinkidMainPatternRecog2 {
    private static final Logger log = LoggerFactory.getLogger(LinkidMainPatternRecog2.class);

    /**
     * 获取包含同样次序linkid序列的tid列表
     * @param map tid及其linkid列表
     * @param roadLinkids 待查的次序linkid列表
     * @return 满足条件的tid列表
     */
    public static List<Integer> getLinkidList(Map<Integer, List<Integer>> map,List<Integer> roadLinkids) {
        // Configure engine with event names to make the statements more readable.
        // This could also be done in a configuration file.
        Configuration configuration = new Configuration();
        configuration.addEventType("TerminalLink", TerminalLink.class.getName());
        // Get engine instance
        EPServiceProvider epService = EPServiceProviderManager.getProvider("linkid", configuration);
        epService.getEPAdministrator().getConfiguration().setMatchRecognizeMaxStates(100L);

        List<Integer> tidList=new ArrayList<Integer>(300000);

//        String pattern = "select * from TerminalLink "
//                + "match_recognize ("
//                + "measures A.tid as tid "
//                + "pattern (A B* C D* E) "
//                + "define "
//                + "A as A.linkid=1,"
//                + "B as B.tid=A.tid,"
//                + "C as C.tid=A.tid and C.linkid=2,"
//                + "D as D.tid=A.tid,"
//                + "E as E.tid=A.tid and E.linkid=3)";

        //produce epl statement
        StringBuffer pattern = new StringBuffer().append("context parttid select * from TerminalLink match_recognize ( measures A.tid as tid after match skip to next row ");
        StringBuffer _sub_pattern = new StringBuffer().append(" pattern ( ");
        StringBuffer _sub_define = new StringBuffer().append(" define ");
        int roadlinkid = 0;
        for(int z=0;z<roadLinkids.size();z++){
            roadlinkid = roadLinkids.get(z);
            if (z==0) {
                _sub_pattern.append(" A ").append(" B0* ");
                _sub_define.append(" A as A.linkid=").append(roadlinkid).append(", B0 as B0.tid=A.tid, ");
            } else if (z==(roadLinkids.size()-1)) {
                _sub_pattern.append(" A").append(z).append(") ");
                _sub_define.append(" A").append(z).append(" as A").append(z).append(".linkid=").append(roadlinkid)
                        .append(" and A").append(z).append(".tid=A.tid)");
            } else {
                _sub_pattern.append(" A").append(z).append(" B").append(z).append("* ");
                _sub_define.append(" A").append(z).append(" as A").append(z).append(".linkid=").append(roadlinkid)
                        .append(" and A").append(z).append(".tid=A.tid, ")
                        .append(" B").append(z).append(" as B").append(z).append(".tid=A.tid, ");
            }
        }

        String epl = pattern.append(_sub_pattern).append(_sub_define).toString();
        log.info("epl:"+epl);
        epService.getEPAdministrator().createEPL("create context parttid partition by tid from TerminalLink");
        EPStatement statement = epService.getEPAdministrator().createEPL(epl);
        LinkidPatternListener2 linkidPatternListener2 = new LinkidPatternListener2();
        statement.addListener(linkidPatternListener2);

        EPRuntime runtime = epService.getEPRuntime();
        for (Map.Entry<Integer, List<Integer>> entry : map.entrySet()) {
//            System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());
//            if (i%1000000==0) {
//                log.info("patterning......:"+i);
//            }
            if (roadLinkids.size()>0) {
                for (int j = 0; j < entry.getValue().size(); j++) {
                    TerminalLink tl = new TerminalLink();
                    tl.setTid(entry.getKey());
                    tl.setLinkid(entry.getValue().get(j));
                    runtime.sendEvent(tl);
                }
                if (linkidPatternListener2.getLastEvent()!=null) {
                    EventBean event = linkidPatternListener2.getLastEvent();
                    tidList.add((int)event.get("tid"));
                }
            }
        }
        epService.destroy();

        //get distinct tid list, and sort them
        List<Integer> unique = tidList.stream().distinct().collect(Collectors.toList());
        Collections.sort(unique);

        return unique;
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        log.info("making test data begin......");
        //创建tid及其linkid列表测试数据集
        Map<Integer, List<Integer>> map = new HashMap<Integer, List<Integer>>(300000);
        for (int i = 700001; i < 901100; i++) {
            List<Integer> li = new ArrayList<Integer>();
            for (int j = 1; j < 240; j++) {
                li.add(j+901099-i);
            }
            map.put(i,li);
        }
        //创建待查的次序linkid列表测试数据集
        List<Integer> roadLinkids=new ArrayList<Integer>();
        for (int j = 1; j < 4; j++) {
            roadLinkids.add(200000-1+j);
        }
        log.info("making test data end......");

        log.info("getting tid list begin ......");
        List<Integer> unique= getLinkidList(map,roadLinkids);
        log.info("getting tid list end ......");

        log.info("printing tid list begin ......");
        Iterator<Integer> it2 = unique.iterator();
        while (it2.hasNext()) {
            log.info(it2.next() + " ");
        }
        log.info("printing tid list end ......");
    }
}
